package com.my.hbase;

import com.my.hbase.conf.Config;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class CRUDTable2 {

    public static void main(String[] args) throws Exception {
        Connection connection = Config.getConnection();
        Admin admin = connection.getAdmin();
        if (admin.getNamespaceDescriptor("myNamespace") == null){
            admin.createNamespace(NamespaceDescriptor.create("myNamespace").build());
            System.out.println("创建namespace成功");
        }
        TableName tableName = TableName.valueOf("myNamespace","Order");
        if (!admin.tableExists(tableName)){
            //构建表
            TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName);
            //构建列族对象
            List<ColumnFamilyDescriptor> families = new ArrayList<ColumnFamilyDescriptor>();
            ColumnFamilyDescriptor id = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("id")).build();
            ColumnFamilyDescriptor name = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("name")).build();
            ColumnFamilyDescriptor amount = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("amount")).build();
            ColumnFamilyDescriptor data = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("data")).build();
            families.add(id);
            families.add(name);
            families.add(amount);
            families.add(data);
            //添加列簇
            tableDescriptor.setColumnFamilies(families);
            //创建表
            admin.createTable(tableDescriptor.build());
            System.out.println("创建表成功");
        }
        //获取Table对象
        Table table = connection.getTable(tableName);
        try {
            put(table);
            get(table);
            scan(table);
        } finally {
            try {
                //使用完了要释放资源
                if (table != null) table.close();
            } catch (Exception e) {
                e.getStackTrace();
            }
        }
    }


    //插入数据
    private static void put(Table table) throws Exception {
        //定义行
        byte[] row = Bytes.toBytes("row1");
        //创建Put对象
        Put put = new Put(row);
        //列簇
        byte[] columnFamily = Bytes.toBytes("name");
        //列
        byte[] qualifier = Bytes.toBytes(String.valueOf(1));
        //值
        byte[] value = Bytes.toBytes("张三丰1");
        //封装Put对象
        put.addColumn(columnFamily, qualifier, value);
        //向表中添加数据
        table.put(put);
        System.out.println("数据插入成功");
    }

    //获取数据
    private static void get(Table table) throws Exception {
        //定义get对象
        Get get = new Get(Bytes.toBytes("row1"));
        //同row多版本数据
        get.setMaxVersions(Integer.MAX_VALUE);
        //通过table对象获取数据
        Result result = table.get(get);
        System.out.println("Result: " + result);
        //只需要获取“值” 这里表示获取 data:1 列族的值
        byte[] valueBytes = result.getValue(Bytes.toBytes("name"), Bytes.toBytes("1")); //获取到的是字节数组
        //将字节转成字符串
        String valueStr = new String(valueBytes, "utf-8");
        System.out.println("value:" + valueStr);
        //多版本数据结果
        for (Cell cell : result.rawCells()) {
            System.out.println("name:"+Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
        }
    }

    //扫描
    private static void scan(Table table) throws Exception {
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        for (Result scannerResult : scanner) {
            System.out.println("数据个数："+scannerResult.size());
            System.out.println("Scan: " + scannerResult);
            for (Cell cell : scannerResult.listCells()) {
                String value = Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength());
                System.out.println(cell + ":" + value);
            }
        }
    }
}
